package com.ls.mq.queue;

import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;

import com.ls.mq.iterface.MessageListener;
import com.ls.mq.listener.impl.DefaultMessageListener;
import com.ls.mq.redis.RedisFactory;
import com.ls.mq.util.Toolkit;

import redis.clients.jedis.Jedis;

public class ListenQueueMessageTask implements Runnable{
	Logger logger = Logger.getLogger(ListenQueueMessageTask.class);
	public static final int timeout=1;//秒
	Jedis client  ;
	/**
	 * channel listener
	 */
	private Map<String, MessageListener> listenerMap;
	private MessageListener defaultListener;
	public ListenQueueMessageTask(Map<String, MessageListener> listenerMap){
		this.listenerMap=listenerMap;
		defaultListener = new DefaultMessageListener();
	}
	
	@Override
	public void run() {
		
		while(true){
			if(client == null)
				client = getConnection();
			Set<String> keys = listenerMap.keySet();
			if(Toolkit.notEmpty(keys)){
				List<String> msg = client.brpop(timeout, keys.toArray(new String[0]));
				dealMsg(msg);
			}else{
				try {
					Thread.sleep(timeout*1000);
				} catch (InterruptedException e) {
					logger.info(e);
				}
			}
			if(Thread.currentThread().isInterrupted()){
				logger.info("任务中断，停止执行");
				break;
			}
			
		}
		
	}

	private Jedis getConnection() {
		return RedisFactory.getInstance().getClient();
	}

	private void dealMsg(List<String> msg) {
		try{
		if(Toolkit.notEmpty(msg)&&msg.size()>1){
			String channel = msg.get(0);
			String jsonStr = msg.get(1);
			MessageListener listener = listenerMap.get(channel);
			if(listener!=null)
				listener.doMessage(channel, jsonStr);
			else
				defaultListener.doMessage(channel, jsonStr);
		}
		}catch(Throwable e){
			e.printStackTrace();
		}
		
	}

}
